package edu.xidian.sei.mlcs

import java.io.File

import scala.collection.mutable.HashMap
import scala.collection.mutable.LinkedHashMap
import scala.collection.mutable.ListBuffer
import scala.concurrent.Await
import scala.concurrent.duration.DurationInt

import com.typesafe.config.ConfigFactory
import akka.pattern.ask
import akka.util.Timeout

import scala.concurrent.Await
import scala.concurrent.duration.DurationInt
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.ActorSystem
import akka.actor.Props
import akka.actor.actorRef2Scala
import akka.util.Timeout
import edu.xidian.sei.mlcs.SubGraphActor.addressMap

class SubGraphActor(id: Int, mlcs: MLCS, bufferSize: Int, system: ActorSystem) extends Actor {
  val subGraphMap = new SubGraphMap
  val bufferMap = new LinkedHashMap[Location, InNode]

  def receive = {

    case InitialNode(loc: Location) => {
      newNode(loc)
    }

    case BuildGraph(loc: Location) => {
      sender ! addEdge(null, loc)
    }

    case ForwardTopoOrdering(from: Location, to: Location, level: Int) => {
      val nodeOpt = getNode(to)
      if (!nodeOpt.isEmpty) nodeOpt.get.addPredecessor(from, level)
      val successors = mlcs.nextLocations(to)
      if (successors.isEmpty) {
        addressMap(mlcs.end.indexSum % addressMap.size) ! ForwardTopoENDNode(to, level + 1)
      } else {
        successors.foreach { loc =>
          val machineID = loc.indexSum % addressMap.size
          addressMap(machineID) ! ForwardTopoOrdering(to, loc, level + 1)
        }
      }
    }

    case ForwardTopoENDNode(to: Location, level: Int) => {
      val nodeOpt = getNode(mlcs.end)
      if (!nodeOpt.isEmpty) nodeOpt.get.addEndPredecessor(to, level)
    }

    case ForwardTopoFinish => {
      val nodeOpt = getNode(mlcs.end)
      if (!nodeOpt.isEmpty) sender ! (nodeOpt.get.indegree == 0)
    }

    case ReverseTopoENDNode => {
      val nodeOpt = getNode(mlcs.end).get
      sender ! Tuple2(nodeOpt.predecessors, nodeOpt.level)
    }

    case ReverseTopoOrdering(to: Location, level: Int, maxLevel: Int, fromCount: BigDecimal) => {
      val nodeOpt = getNode(to)
      if (!nodeOpt.isEmpty) {
        val node = nodeOpt.get
        if (node.level + level == maxLevel) {
          node.addCount(fromCount)
          sender ! Tuple3(node.predecessors, to, node.count)
        } else {
          sender ! Tuple3(new ListBuffer[Location](), null, null)
        }
      } else {
        sender ! Tuple3(new ListBuffer[Location](), null, null)
      }
    }

    case ReverseTopoFinish => {
      val nodeOpt = getNode(mlcs.start).get
      sender ! nodeOpt.count
    }

    case "finish" =>
      println("finish")
      context.stop(self)
      system.terminate()

  }

  def getNode(loc: Location): Option[InNode] = {
    if (bufferMap.contains(loc)) {
      bufferMap.get(loc)
    } else if (fileExist(loc)) {
      val node = subGraphMap.nodeFileReader(loc)
      dequeue
      bufferMap.put(loc, node)
    } else {
      None
    }
  }

  def newNode(loc: Location): InNode = {
    val node = new InNode(loc)
    dequeue
    bufferMap.put(loc, node)
    node
  }

  def addEdge(start: Location, to: Location): Boolean = {
    val n = getNode(to)
    if (n.isEmpty) {
      val node = newNode(to)
      node.inc
      true
    } else {
      val node = n.get
      node.inc
      false
    }
  }

  def dequeue() {
    if (bufferMap.size > bufferSize) {
      val pair = bufferMap.head
      subGraphMap.nodeFileWriter(pair._2)
      bufferMap.remove(pair._1)
    }
  }

  def fileExist(loc: Location): Boolean = {
    val file = new File(loc.toFileIndex + ".txt")
    file.exists()
  }

}

object SubGraphActor {
  implicit val timeout = Timeout(10 seconds)
  val addressMap = new HashMap[Int, ActorRef]
  val system = ActorSystem("SubGraph", ConfigFactory.parseFile(new File("subgraph.conf")).getConfig("subgraphactor"))

  def addAddress(adds: Array[String]) {
    adds.foreach { add =>
      val strs = add.split("@")
      val otherActor = Await.result(system.actorSelection(s"akka.tcp://SubGraph@${strs(1)}/user/subgraph").resolveOne(), timeout.duration)
      addressMap.put(strs(0).toInt, otherActor)
    }
  }

  def main(args: Array[String]): Unit = {

    val config = ConfigFactory.parseFile(new File("subgraph.conf")).getConfig("subgraphactor")
    val id = config.getInt("params.id")
    val addresses = config.getString("params.addresses").split("|")
    addAddress(addresses)
    val mlcs = MLCS.loadFromFile(config.getString("params.inputFile"))
    val bufferSize = config.getInt("params.bufferSize")
    system.actorOf(Props(classOf[SubGraphActor], id, mlcs, bufferSize, system), name = "subgraph")

  }
}

case class InitialNode(loc: Location)

case class BuildGraph(loc: Location)

case class ForwardTopoOrdering(from: Location, to: Location, level: Int)

case class ForwardTopoENDNode(to: Location, level: Int)

case class ForwardTopoFinish()

case class ReverseTopoENDNode()

case class ReverseTopoOrdering(to: Location, level: Int, maxLevel: Int, fromCount: BigDecimal)

case class ReverseTopoFinish()